package sc.writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;

/**
 * A writable type for MapReduce application. It stores an extensible array.
 * 
 * @author Quentin Rossé
 *
 * @param <E> Writable type of the data stored in the array
 */
public abstract class WritableArrayList<E extends Writable> implements Writable{

	private static final int DEFAULT_SIZE = 16;

	private static final int INCREASE_SIZE_RATE = 2;
	
	private static final Configuration CONF = new Configuration();
	
	private Writable[] array;
	
	/**
	 * start indice. Only used when removing elements
	 */
	private int start = 0;
	
	private int size = 0;
	
	private Class<? extends Writable> valueClass;
	
	protected WritableArrayList(Class<? extends Writable> valueClass) {
		this.valueClass = valueClass;
		size = 0;
		
		resizeArray(DEFAULT_SIZE);
	}
	
	/**
	 * add an item to the end of the list
	 * 
	 * @param item object to add
	 */
	public void add(E item){
		if(size + start >= array.length)
			resizeArray(size * INCREASE_SIZE_RATE);
		array[start + size++] = item;
	}
	
	/**
	 * add an item to the end of the list. If the list has more than maxsize
	 * elements, the fist(s) will be lost 
	 * 
	 * @param item object to add
	 */
	public void add(E item, int maxsize){
		if(size + start >= array.length)
			resizeArray(size * INCREASE_SIZE_RATE);
		array[start + size++] = item;
		if(size > maxsize){
			start += size - maxsize;
			size = maxsize;
		}
	}
	
	@SuppressWarnings("unchecked")
	public E get(int i){
		return (E) array[i];
	}
	
	public int size(){
		return size;
	}
	
	public void clear(){
		start = 0;
		size = 0;
	}
	
	@Override
	public void readFields(DataInput in) throws IOException {
		clear();
		
		//we don't update the size attribute right away to do the resize
		//correctly if one is needed
		int size = in.readInt();
		
		if(size > array.length)
			resizeArray(size);
		for(int i=0; i<size; i++){
			array[i].readFields(in);
		}
		
		this.size = size;
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeInt(size);
		for(int i=start; i<size+start; i++){
			array[i].write(out);
		}
	}
	
	@Override
	public String toString(){
		String s = "[";
		for(int i=start; i<size+start; i++){
			s += array[i].toString() + ", ";
		}
		int lastComa = s.lastIndexOf(',');
		if(lastComa < 0)
			lastComa = 1;
		return s.substring(0, lastComa) + "]";
	}
	
	
	private void resizeArray(int newSize){
		System.out.println("old size : " + size + "  new size : " +newSize);
		Writable[] tmpArray = new Writable[newSize];
		for(int i=0; i<tmpArray.length; i++){
			if(i>=start && i+start < size)
				tmpArray[i] = array[i+start];
			else{
				tmpArray[i] = ReflectionUtils.newInstance(valueClass, CONF);
			}
		}
		array = tmpArray;
	}

}
